package MapReduce.Demo3_InputFormat;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import utils.JobSubmit;

import java.io.IOException;

/**
 * @Author lixin
 * @Date 2023/3/15 10:56
 */
public class KeyValueFileInputFormatApp {

    public static class KVFIFMapper extends Mapper<Text,Text,Text, Text>{

        @Override
        protected void map(Text year, Text temperature, Mapper<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {

            context.write(year,temperature);

        }
    }

    public static class KVFIFReducer extends Reducer<Text,Text,Text,Text> {

        Text outKey = new Text();
        Text outValue = new Text();

        @Override
        protected void reduce(Text key, Iterable<Text> values, Reducer<Text, Text, Text, Text>.Context context) throws IOException, InterruptedException {

            for (Text value : values) {
                outKey.set("key:"+key);
                outValue.set("value:"+value);
                context.write(outKey,outValue);
            }

        }
    }


    public static void main(String[] args) throws Exception {

//        Configuration conf = new Configuration();
//
//        Job job = Job.getInstance(conf);
//
//        conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPARATOR,",");
//
//        job.setMapOutputKeyClass(IntWritable.class);
//        job.setMapOutputValueClass(IntWritable.class);
//        job.setOutputKeyClass(IntWritable.class);
//        job.setOutputValueClass(IntWritable.class);
//
//        job.setJarByClass(SequenceFileInputFormatApp.class);
//        job.setMapperClass(SFIFMapper.class);
//        job.setReducerClass(SFIFReducer.class);
//
//
//        Path outputPath = new Path(args[1]);
//
//        FileSystem fs = FileSystem.get(conf);
//
//        if (fs.exists(outputPath)) {
//            fs.delete(outputPath,true);
//        }
//
//        //设置输入格式化的类型，如果不设置，模式任用TextInputFormat
//        job.setInputFormatClass(SequenceFileInputFormat.class);
//
//        FileInputFormat.setInputPaths(job,new Path(args[0]));
//        FileOutputFormat.setOutputPath(job,outputPath);
//
//        job.waitForCompletion(true);

        JobSubmit.setSeperator(",");
        JobSubmit.submitBaseJob(KeyValueFileInputFormatApp.class,args, KeyValueTextInputFormat.class);

    }

}
